package region_distribution

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by yangqiyuan on 2018/3/28.
  */
object RegionDistribution {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
    .setAppName(RegionDistribution.getClass.getSimpleName)
    .setMaster("local[*]")

    val spark: SparkContext = new SparkContext(conf)
    val sqlContext: SQLContext = new SQLContext(spark)
    val parquetFile: DataFrame = sqlContext.read.parquet("parquet")
    var init_request = 0
    var effective_request = 0
    var ad_request =0
    var bid_count =0
    var sucess_bid_count =0
    var show_count =0
    var click_count =0
    var ad_cost =0
    var ad_consume =0
    val regionRdd: RDD[((String, String), List[Int])] = parquetFile.map(line => {
      //取出需要用到的字段
      val provincename: String = line.getAs("provincename")
      val cityname: String = line.getAs("cityname")
      val requestmode: Int = line.getAs("requestmode")
      val processnode: Int = line.getAs("processnode")
      val iseffective: Int = line.getAs("iseffective")
      val isbilling: Int = line.getAs("isbilling")
      val isbid: Int = line.getAs("isbid")
      val iswin: Int = line.getAs("iswin")
      val adorderid: Int = line.getAs("adorderid")


      //对取出来的数据进行判断
      init_request = if (requestmode == 1 && processnode >= 1) 1 else 0

      effective_request = if (requestmode == 1 && processnode >= 2) 1 else 0

      ad_request = if (requestmode == 1 && processnode == 3) 1 else 0

      bid_count = if (iseffective.equals(1) && isbilling.equals(1) && isbid.equals(1) && adorderid != 0) 1 else 0

      sucess_bid_count = if (iseffective == 1 && isbilling == 1 && iswin == 1) 1 else 0

      show_count = if (requestmode == 2 && iseffective == 1) 1 else 0

      click_count = if (requestmode == 3 && iseffective == 1) 1 else 0

      ad_cost = if (iseffective == 1 && isbilling == 1 && iswin == 1) 1 else 0

      ad_consume = if (iseffective == 1 && isbilling == 1 && iswin == 1) 1 else 0

      ((provincename, cityname), List[Int](init_request, effective_request, ad_request, bid_count, sucess_bid_count, show_count, click_count, ad_cost, ad_consume))
      /*else if(requestmode==1&&processnode>=2){
       effective_request +=1
     }else if(requestmode==1&&processnode==3){
       ad_request +=1
     }else if(iseffective.equals(1)&&isbilling.equals(1)&&isbid.equals(1)&&adorderid!=0){
       bid_count +=1
     }else if(iseffective==1&&isbilling==1&&iswin==1){
       sucess_bid_count +=1
     }else if(requestmode==2&&iseffective==1){
       show_count +=1
     }else if(requestmode==3&&iseffective==1){
       click_count +=1
     }else if(iseffective==1&&isbilling==1&&iswin==1){
       ad_cost +=1
     }else if(iseffective==1&&isbilling==1&&iswin==1){
       ad_consume +=1
     }*/


    })
    //对每个省每个市的数据进行分组聚合
    val result: RDD[((String, String), List[Int])] = regionRdd.reduceByKey((list1, list2) => {
      list1.zip(list2).map(t => t._1 + t._2)
    })




  }
}
